package kafkastudy.concurrent;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * Created by hp on 2018/9/14.
 */
public class ConsumerSimple {

    public static void main(String[] args) {

        KafkaConsumer<String, String> consumer = null;
        try {
            Properties props = new Properties();
            props.put("bootstrap.servers", "10.1.1.42:9092,10.1.1.43:9092,10.1.1.44:9092");
            props.put("group.id", "group17");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            props.put("auto.offset.reset", "earliest");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList("logrule"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records)
                    System.out.println("offset = "+record.offset()+", key =" +record.key()+", value ="+record.value());
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(consumer != null) {
                consumer.close();
            }
        }
    }
}
